package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Demo2AggregateByKey {
  def main(args: Array[String]): Unit = {
    //使用reduceByKey/aggregateByKey替代groupByKey
    val sparkSession: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions", "1")
      .master("local[2]")
      .appName("缓存优化")
      .getOrCreate()

    val sparkContext: SparkContext = sparkSession.sparkContext


    val stuRDD: RDD[String] = sparkContext.textFile("spark/data/students.txt")

    val clazzKVRDD: RDD[(String, Int)] = stuRDD.map(_.split(",") match {
      case Array(_, _, _, _, clazz: String) =>
        (clazz, 1)
    })

    //reduceByKey的使用，分组之后，直接使用聚合
    //理解为普通的MapReduce中的根据相同的键进入到同一个reduce, 然后在reduce端聚合
    //实际上这里对应的是前一个RDD中的分区中数据相同的键到后一个RDD中同一个分区，在后一个RDD分区中的聚合
    //    val resRDD: RDD[(String, Int)] = clazzKVRDD.reduceByKey(_ + _)
    //    resRDD.foreach(println)

    //groupByKey 不做聚合，只做前一个RDD中的分区中数据相同的键到后一个RDD中同一个分区 (尽量使用reduceByKey去替代)
    //    val resRDD2: RDD[(String, Int)] = clazzKVRDD.groupByKey()
    //      .map((kv: (String, Iterable[Int])) => {
    //        (kv._1, kv._2.sum)
    //      })
    //    resRDD2.foreach(println)

    //aggregateByKey
    //aggregateByKey(zeroValue: U)(seqOp: (U, V) => U,  combOp: (U, U) => U)
    //zeroValue: 初始值，这个参数只会被后面第一个参数函数所使用
    //seqOp: 相当于map端预聚合的逻辑
    //combOp: 相当于reduce端的聚合逻辑
    val resRDD3: RDD[(String, Int)] = clazzKVRDD.aggregateByKey(0)(
      //相当于map端预聚合的逻辑
      (a1: Int, a2: Int) => a1 + a2,
      //相当于reduce端的聚合逻辑
      (b1: Int, b2: Int) => b1 + b2
    )

    resRDD3.foreach(println)


  }
}
